package com.ccc.demo1;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.graphx.Edge;
import org.apache.spark.graphx.EdgeDirection;
import org.apache.spark.graphx.Graph;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;
import scala.reflect.ClassTag;

import java.util.ArrayList;
import java.util.List;

public class BlackListJob {

    public static void main(String[] args) throws Exception{
        SparkConf sparkConf = new SparkConf().setAppName("demo").setMaster("local[2]");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

        List<Tuple2<Object, String>> listOfVertex = new ArrayList<>();
        listOfVertex.add(new Tuple2<>(1l, "black"));
        listOfVertex.add(new Tuple2<>(2l, "white"));
        listOfVertex.add(new Tuple2<>(3l, "white"));
        listOfVertex.add(new Tuple2<>(4l, "white"));
        listOfVertex.add(new Tuple2<>(5l, "black"));
        listOfVertex.add(new Tuple2<>(6l, "white"));

        List<Edge<String>> listOfEdge = new ArrayList<>();
        listOfEdge.add(new Edge<>(1, 3, ""));
        listOfEdge.add(new Edge<>(2, 3, ""));
        listOfEdge.add(new Edge<>(4, 3, ""));
        listOfEdge.add(new Edge<>(3, 6, ""));
        listOfEdge.add(new Edge<>(5, 6, ""));

        JavaRDD<Tuple2<Object, String>> vertexRDD = javaSparkContext.parallelize(listOfVertex);
        JavaRDD<Edge<String>> edgeRDD = javaSparkContext.parallelize(listOfEdge);

        ClassTag<String> stringTag = scala.reflect.ClassTag$.MODULE$.apply(String.class);

        Graph<String, String> graph = Graph.apply(
                vertexRDD.rdd(),
                edgeRDD.rdd(),
                null,
                StorageLevel.MEMORY_ONLY(),
                StorageLevel.MEMORY_ONLY(),
                stringTag,
                stringTag
        );

        graph = graph.ops().pregel(
                "white",         // 初始消息
                2,            // 最大轮次：第一轮是默认逻辑
                EdgeDirection.Both(),     // 双向
                new MyVprog(),            // 根据消息更新节点状态
                new MySendMsg(),          // 发消息的逻辑
                new MyMergeMsg(),         // 消息归并的逻辑
                stringTag                 // 指明消息对象内容
        );

        graph.vertices().toJavaRDD().foreach(x -> System.out.println(x));
    }
}
